package com.neuedu.myweather;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class Run {
    public static void show(Path path) {
        try {
            for (FileStatus f : HadoopUtils.getFs().listStatus(path)) {
                if (f.isFile()) {
                    FSDataInputStream inputStream = HadoopUtils.getFs().open(f.getPath());
                    BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
                    String line = reader.readLine();
                    while (line != null) {
                        System.out.println(line);
                        line = reader.readLine();
                    }
                    reader.close();
                    inputStream.close();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        try {
            Configuration conf = new Configuration();
            FileSystem hdfs = FileSystem.get(conf);
            String input = "/myweather";
            String output = "/myweather_output";
            Path outputPath = new Path(output);
            // 判断输出目录是否存在，存在则删除之
            HadoopUtils.delete(outputPath);
            // 创建Job任务
            Job job = Job.getInstance(conf, "每年最高温度算法");
            //设置输入
            job.setInputFormatClass(TextInputFormat.class);
            FileInputFormat.setInputPaths(job, input);
            // 设置mapper
            job.setMapperClass(WeatherMapper.class);
            job.setMapOutputKeyClass(WeatherWritable.class);
            job.setMapOutputValueClass(Text.class);
            //设置分区
            job.setNumReduceTasks(3);
            job.setPartitionerClass(YearPartitioner.class);
            //自定义排序
            job.setSortComparatorClass(TemperatureOfDescComparator.class);
            // 设置分组
            job.setGroupingComparatorClass(YearGrouping.class);
            //设置reducer
            job.setReducerClass(WeatherReducer.class);
            // 设置输出:下面2行容易写错
            job.setOutputKeyClass(WeatherWritable.class);
            job.setOutputValueClass(NullWritable.class);
            job.setOutputFormatClass(TextOutputFormat.class);
            FileOutputFormat.setOutputPath(job, outputPath);
            // 运行
            boolean flag = job.waitForCompletion(true);
            if (flag) {
                System.out.println("每年最高温度算法，结束~~~");
                show(outputPath);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
